/**
 * Copyright (c) 2013-2022 Nikita Koksharov
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.redisson;

import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import org.redisson.api.RFuture;
import org.redisson.api.RLock;
import org.redisson.client.RedisException;
import org.redisson.client.codec.Codec;
import org.redisson.client.codec.LongCodec;
import org.redisson.client.protocol.RedisCommand;
import org.redisson.client.protocol.RedisCommands;
import org.redisson.client.protocol.convertor.IntegerReplayConvertor;
import org.redisson.client.protocol.decoder.MapValueDecoder;
import org.redisson.command.CommandAsyncExecutor;
import org.redisson.misc.CompletableFutureWrapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;

/**
 * 用于实现分布式锁的基类
 */
public abstract class RedissonBaseLock extends RedissonExpirable implements RLock {


    /**
     * 描述带有过期时间的键值对的数据结构。
     * 它包含了键值对以及过期时间的信息。
     * ExpirationEntry通常用于缓存和数据库中的键值对，以实现自动过期和时间限制的功能。
     */
    public static class ExpirationEntry {

        private final Map<Long, Integer> threadIds = new LinkedHashMap<>();
        private volatile Timeout timeout;

        public ExpirationEntry() {
            super();
        }

        /**
         * 添加
         */
        public synchronized void addThreadId(long threadId) {
            threadIds.compute(threadId, (t, counter) -> {
                counter = Optional.ofNullable(counter).orElse(0);
                counter++;
                return counter;
            });
        }


        /**
         * 判断
         */
        public synchronized boolean hasNoThreads() {
            return threadIds.isEmpty();
        }


        /**
         * 获取
         */
        public synchronized Long getFirstThreadId() {
            if (threadIds.isEmpty()) {
                return null;
            }
            return threadIds.keySet().iterator().next();
        }

        /**
         * 删除
         */
        public synchronized void removeThreadId(long threadId) {
            threadIds.compute(threadId, (t, counter) -> {
                if (counter == null) {
                    return null;
                }
                counter--;
                if (counter == 0) {
                    return null;
                }
                return counter;
            });
        }

        public void setTimeout(Timeout timeout) {
            this.timeout = timeout;
        }

        public Timeout getTimeout() {
            return timeout;
        }

    }


    private static final Logger log = LoggerFactory.getLogger(RedissonBaseLock.class);

    /**
     * 过期续订
     */
    private static final ConcurrentMap<String, ExpirationEntry> EXPIRATION_RENEWAL_MAP = new ConcurrentHashMap<>();
    protected long internalLockLeaseTime;

    final String id;
    final String entryName;

    public RedissonBaseLock(CommandAsyncExecutor commandExecutor, String name) {
        super(commandExecutor, name);
        this.id = getServiceManager().getId();
        this.internalLockLeaseTime = getServiceManager().getCfg().getLockWatchdogTimeout();
        this.entryName = id + ":" + name;
    }


    protected String getEntryName() {
        return entryName;
    }


    // 获取锁名称
    protected String getLockName(long threadId) {
        return id + ":" + threadId;
    }


    /**
     * 更新锁的过期时间。 这个方法会延长锁的生存期，使其保持锁定状态更长时间。
     * 会把时间三等分
     * 当调用该方法，Redission将检查当前锁的剩余过期时间;
     * 如果剩余时间小于等于0，即锁已经过期或已经解锁,那么redission将不会做任何事情.
     * 如果剩余时间大于0， 那么Redission将更新锁的过去时间，使其在新的时间点过期
     */
    private void renewExpiration() {
        ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
        if (ee == null) {
            return;
        }

        Timeout task = getServiceManager().newTimeout(new TimerTask() {
            @Override
            public void run(Timeout timeout) throws Exception {
                ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());
                if (ent == null) {
                    return;
                }
                Long threadId = ent.getFirstThreadId();
                if (threadId == null) {
                    return;
                }

                CompletionStage<Boolean> future = renewExpirationAsync(threadId);
                future.whenComplete((res, e) -> {
                    if (e != null) {
                        log.error("Can't update lock {} expiration", getRawName(), e);
                        EXPIRATION_RENEWAL_MAP.remove(getEntryName());
                        return;
                    }
                    if (res) {
                        // reschedule itself
                        renewExpiration();
                    } else {
                        cancelExpirationRenewal(null);
                    }
                });
            }
        }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);

        ee.setTimeout(task);
    }





    /**
     * 安排过期续期。
     * 此方法用于在映射中添加或更新一个线程ID，以表示该线程正在处理某个资源的过期续期。
     * 如果该资源的过期续期条目已存在，则将当前线程ID添加到已存在的条目中。
     * 如果该资源的过期续期条目不存在，则创建一个新的条目，并尝试续期过期时间。
     * 在尝试续期过程中，如果当前线程被中断，则取消该资源的过期续期。
     */
    protected void scheduleExpirationRenewal(long threadId) {
        // 创建一个新的过期续期条目。
        ExpirationEntry entry = new ExpirationEntry();
        // 尝试将新条目添加到映射中，如果已存在则获取现有的条目。
        ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);
        // 如果现有条目存在，则将当前线程ID添加到条目中。
        if (oldEntry != null) {
            oldEntry.addThreadId(threadId);
        } else {
            // 如果现有条目不存在，则将当前线程ID添加到新条目中，并尝试续期。
            entry.addThreadId(threadId);
            try {
                renewExpiration();
            } finally {
                // 在续期操作完成后，检查当前线程是否被中断。
                // 如果是，则取消当前线程的过期续期操作。
                if (Thread.currentThread().isInterrupted()) {
                    cancelExpirationRenewal(threadId);
                }
            }
        }
    }


    protected CompletionStage<Boolean> renewExpirationAsync(long threadId) {
        return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                        "return 1; " +
                        "end; " +
                        "return 0;",
                Collections.singletonList(getRawName()),
                internalLockLeaseTime, getLockName(threadId));
    }



    protected void cancelExpirationRenewal(Long threadId) {
        ExpirationEntry task = EXPIRATION_RENEWAL_MAP.get(getEntryName());
        if (task == null) {
            return;
        }

        if (threadId != null) {
            task.removeThreadId(threadId);
        }

        if (threadId == null || task.hasNoThreads()) {
            Timeout timeout = task.getTimeout();
            if (timeout != null) {
                timeout.cancel();
            }
            EXPIRATION_RENEWAL_MAP.remove(getEntryName());
        }
    }

    protected final <T> RFuture<T> evalWriteAsync(String key, Codec codec, RedisCommand<T> evalCommandType, String script, List<Object> keys, Object... params) {
        return commandExecutor.syncedEval(key, codec, evalCommandType, script, keys, params);
    }

    protected void acquireFailed(long waitTime, TimeUnit unit, long threadId) {
        commandExecutor.get(acquireFailedAsync(waitTime, unit, threadId));
    }

    /**
     * 抽象类方法,可重写
     * 只有RedissonFairLock重写了这方法
     */
    protected CompletableFuture<Void> acquireFailedAsync(long waitTime, TimeUnit unit, long threadId) {
        return CompletableFuture.completedFuture(null);
    }


    protected void trySuccessFalse(long currentThreadId, CompletableFuture<Boolean> result) {
        acquireFailedAsync(-1, null, currentThreadId).whenComplete((res, e) -> {
            if (e == null) {
                result.complete(false);
            } else {
                result.completeExceptionally(e);
            }
        });
    }



    @Override
    public Condition newCondition() {
        // TODO implement
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean isLocked() {
        return isExists();
    }

    @Override
    public RFuture<Boolean> isLockedAsync() {
        return isExistsAsync();
    }

    @Override
    public boolean isHeldByCurrentThread() {
        return isHeldByThread(Thread.currentThread().getId());
    }

    @Override
    public boolean isHeldByThread(long threadId) {
        RFuture<Boolean> future = commandExecutor.writeAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.HEXISTS, getRawName(), getLockName(threadId));
        return get(future);
    }

    private static final RedisCommand<Integer> HGET = new RedisCommand<Integer>("HGET", new MapValueDecoder(), new IntegerReplayConvertor(0));

    public RFuture<Integer> getHoldCountAsync() {
        return commandExecutor.writeAsync(getRawName(), LongCodec.INSTANCE, HGET, getRawName(), getLockName(Thread.currentThread().getId()));
    }

    @Override
    public int getHoldCount() {
        return get(getHoldCountAsync());
    }

    @Override
    public RFuture<Boolean> deleteAsync() {
        return forceUnlockAsync();
    }

    @Override
    public RFuture<Void> unlockAsync() {
        long threadId = Thread.currentThread().getId();
        return unlockAsync(threadId);
    }

    @Override
    public RFuture<Void> unlockAsync(long threadId) {
        return getServiceManager().execute(() -> unlockAsync0(threadId));
    }

    /**
     * 异步解锁操作的私有方法。
     * 该方法首先尝试执行内部的异步解锁逻辑，然后根据操作结果进行后续处理，如取消续期任务、抛出异常等。
     * @param threadId 线程ID，标识进行解锁操作的线程。
     * @return 返回一个未来对象（RFuture），该对象封装了异步解锁操作的结果。
     */
    private RFuture<Void> unlockAsync0(long threadId) {
        // 尝试执行内部异步解锁操作。
        CompletionStage<Boolean> future = unlockInnerAsync(threadId);

        // 处理unlockInnerAsync方法返回的结果或异常。
        CompletionStage<Void> f = future.handle((opStatus, e) -> {
            // 不论操作成功与否，都取消与该锁相关的续期任务。
            cancelExpirationRenewal(threadId);

            // 如果操作过程中发生异常，重新抛出CompletionException。
            if (e != null) {
                if (e instanceof CompletionException) {
                    throw (CompletionException) e;
                }
                throw new CompletionException(e);
            }

            // 如果操作状态为null，抛出IllegalMonitorStateException异常，表示尝试解锁的线程并未持有锁。
            if (opStatus == null) {
                IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
                        + id + " thread-id: " + threadId);
                throw new CompletionException(cause);
            }

            // 操作成功，返回null。
            return null;
        });

        // 将CompletionStage包装成RFuture返回。
        return new CompletableFutureWrapper<>(f);
    }


    /**
     * 尝试解锁当前线程持有的锁。
     * 该方法首先会尝试异步解锁，并通过当前线程的ID来标识需要解锁的锁。
     * 如果解锁操作失败，并且失败的原因是非法监视状态异常（IllegalMonitorStateException），
     * 则将该异常抛出；如果是其他类型的Redis异常，则直接抛出该异常。
     *
     * @throws IllegalMonitorStateException 如果解锁时检测到当前线程并不持有锁，则抛出此异常。
     * @throws RedisException 如果解锁操作失败且原因不是非法监视状态异常，则抛出此异常。
     */
    @Override
    public void unlock() {
        try {
            // 异步解锁操作，并通过当前线程ID获取解锁结果
            get(unlockAsync(Thread.currentThread().getId()));
        } catch (RedisException e) {
            // 判断异常类型，对非法监视状态异常进行特殊处理并抛出
            if (e.getCause() instanceof IllegalMonitorStateException) {
                throw (IllegalMonitorStateException) e.getCause();
            } else {
                throw e;
            }
        }
    }

    @Override
    public boolean forceUnlock() {
        return get(forceUnlockAsync());
    }

    protected abstract RFuture<Boolean> unlockInnerAsync(long threadId);

    @Override
    public RFuture<Void> lockAsync() {
        return lockAsync(-1, null);
    }

    @Override
    public RFuture<Void> lockAsync(long leaseTime, TimeUnit unit) {
        long currentThreadId = Thread.currentThread().getId();
        return lockAsync(leaseTime, unit, currentThreadId);
    }

    @Override
    public RFuture<Void> lockAsync(long currentThreadId) {
        return lockAsync(-1, null, currentThreadId);
    }

    @Override
    public RFuture<Boolean> tryLockAsync() {
        return tryLockAsync(Thread.currentThread().getId());
    }

    @Override
    public RFuture<Boolean> tryLockAsync(long waitTime, TimeUnit unit) {
        return tryLockAsync(waitTime, -1, unit);
    }

    @Override
    public RFuture<Boolean> tryLockAsync(long waitTime, long leaseTime, TimeUnit unit) {
        long currentThreadId = Thread.currentThread().getId();
        return tryLockAsync(waitTime, leaseTime, unit, currentThreadId);
    }

    protected final <T> CompletionStage<T> handleNoSync(long threadId, CompletionStage<T> ttlRemainingFuture) {
        return commandExecutor.handleNoSync(ttlRemainingFuture, () -> unlockInnerAsync(threadId));
    }

}
